#include "event.h"
#include "reactor.h"
#include "socket_base.h"
#include "block_queue.h"
#include <sys/socket.h>
#include <errno.h>
#include <netinet/in.h>
#include <fcntl.h>
#include <cstring>

constexpr int UNSUCCESS = -1;
constexpr int SUCCESS = 0;


void SetNonblocking(int sock)
{
    if(sock > 0) {
        int flags = 0;
        flags = fcntl(sock, F_GETFL, 0);
        fcntl(sock, F_SETFL, flags | O_NONBLOCK);
    }
}

void SetAddressReuse(int sock)
{
    int flag = 1;
    int ret = setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, &flag, sizeof(int));
    if(UNSUCCESS == ret) {
        printf("Failed to reuse address\n");
    }
}

Acceptor::Acceptor(SocketReactor* reactor, unsigned short p, const std::string& label)
    : Event(reactor), _label(label)
{
    int fd = socket(AF_INET, SOCK_STREAM, 0);
    SetNonblocking(fd);
    SetAddressReuse(fd);
    _sock = new Socket(fd);

    init(p);
}

void Acceptor::Handle()
{
    sockaddr_in client;
    socklen_t len = sizeof(client);
    int nsock = accept(_sock->Id(), (sockaddr*)&client, &len);

    if(nsock > 0) {
        SetNonblocking(nsock);
        auto reader = new Reader(_reactor, new Socket(nsock));
        reader->SetLabel(_label);
        _reactor->Regist(reader);
    } else {
        printf("Accept failed\n");
    }
}

void Acceptor::init(unsigned short p)
{
    sockaddr_in in;
    
    memset(&in, 0, sizeof(sockaddr_in));
    in.sin_addr.s_addr = INADDR_ANY;
    in.sin_family = AF_INET;
    in.sin_port = htons(p);

    int ret = bind(_sock->Id(), (sockaddr*)&in, sizeof(sockaddr_in));
    if(UNSUCCESS == ret) {
        exit(ret);
    }

    ret = listen(_sock->Id(), 10);
    if(UNSUCCESS == ret) {
        exit(ret);
    }

    printf("Init acceptor successful!\n");
}

void Reader::renew()
{
    if(!_buffer) {
        long long len = 0;
        int nr = recv(_sock->Id(), &len, sizeof(long long), 0);
        if(UNSUCCESS == nr) {
            printf("Error read message length\n");
            return;
        }
        _buffer = new Buffer();
        _buffer->_length = len;
        _buffer->_cbuff = SharedBuffer(new CharBuffer(len));
    }
}

Reader::Reader(SocketReactor *reactor, const std::string &ip, unsigned short p)
    : Reader(reactor, SocketMGR::NewConnect(ip, p))
{
}

void Reader::Handle()
{
    renew();
    if(!_buffer)
        return;

    do
    {
        char* data = _buffer->_cbuff->Get();
        int rlen = recv(_sock->Id(), data + _buffer->_recv, _buffer->_length - _buffer->_recv, 0);
        
        if(UNSUCCESS == rlen) {
            //出错或等待
            if(errno == EWOULDBLOCK || errno == EAGAIN)
                continue;
            else 
                break;
        } else if(0 == rlen) 
            break;

        _buffer->_recv += rlen;
    } while (_buffer->_recv < _buffer->_length);
    
    if(_buffer->_recv == _buffer->_length) {
        ///通过全局的队列管理器，将_buffer->_cbuff放入label指定的队列中
        ///....
        auto* que = QueueRepo::Repo()->Queue<SharedBuffer>(_label);
        que->Push(_buffer->_cbuff);
        ///仅删除_buffer而不能删除_buffer->_cbuff
        delete _buffer;
        _buffer = nullptr;
    }
}
